package com.heima.behavior.service.Impl;


import com.alibaba.fastjson.JSONObject;
import com.heima.behavior.service.ReadBehaviorService;

import com.heima.behavior.pojos.ReadBehavior;
import com.heima.common.constants.ReadBehaviorConstants;
import com.heima.model.behavior.dtos.Behavior;
import com.heima.model.behavior.dtos.ReadBehaviorDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.ArrayList;


@Service
@Slf4j
public class ReadBehaviorServiceImpl implements ReadBehaviorService {

    @Autowired
    private MongoTemplate mongoTemplate;

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;



    @Override
    public Behavior readBehavior(ReadBehaviorDto dto) {
        ReadBehavior readBehavior = new ReadBehavior();
        BeanUtils.copyProperties(dto,readBehavior);
        //查询
        Query query = Query.query(Criteria.where("articleId").is(dto.getArticleId()));
        ReadBehavior one = mongoTemplate.findOne(query, ReadBehavior.class);
        if(one==null){
            mongoTemplate.save(readBehavior);
        }
        //更新
        if(one!=null){
            one.setCount(one.getCount()+1);
            mongoTemplate.save(one);
        }

        //返回数据
        Behavior behavior = new Behavior();

        behavior.setCount(one.getCount());
        behavior.setArticleId(one.getArticleId());

        kafkaTemplate.send(ReadBehaviorConstants.BEHAVIOR_TABLE_NAME, JSONObject.toJSONString(behavior));

        return behavior;

    }
}
